package com.biodiv.community.config;

import com.biodiv.community.enums.TopicEnum;
import com.biodiv.community.manager.mq.RedisListener;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.util.ErrorHandler;

import java.util.*;

/**
 * @author tianquan
 */
@Slf4j
@Configuration
public class RedisConfig implements BeanPostProcessor {

    @Autowired
    private ErrorHandler errorHandler;

    private final Map<String, RedisListener<?>> subscriberMap = new HashMap<>();

    @Override
    public Object postProcessBeforeInitialization(@NotNull Object bean, @NotNull String beanName) throws BeansException {
        if (bean instanceof RedisListener<?> redisListener) {
            TopicEnum topic = redisListener.getTopic();
            subscriberMap.put(topic.getType(), redisListener);
        }
        return bean;
    }

    @Bean
    public RedisMessageListenerContainer redisContainer(RedisConnectionFactory connectionFactory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        container.setErrorHandler(errorHandler);

        // 为不同主题注册监听器
        List<String> topics = new ArrayList<>();
        for (Map.Entry<String, RedisListener<?>> stringAbstractRedisSubscriberEntry : subscriberMap.entrySet()) {
            topics.add(stringAbstractRedisSubscriberEntry.getKey());
            container.addMessageListener(
                    new MessageListenerAdapter(stringAbstractRedisSubscriberEntry.getValue()),
                    ChannelTopic.of(stringAbstractRedisSubscriberEntry.getKey())
            );
        }
        log.info("Registered topics: {}", topics);
        return container;
    }

}